package com.huyi.common.listener;

import java.util.List;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

/**
 * 消费消息
 * 
 * @author shengli.hu
 */
@Component
public class KafkaMessageListener {

    private final Logger logger = LoggerFactory.getLogger(this.getClass());

/*    @KafkaListener(id="myContainer", topics = {"${kafka.consumer.topic}"})
    public void consumerListener(ConsumerRecord<?, ?> record, Acknowledgment ack) {
        logger.info("【myContainer】kafka consumer消费成功！topic=" + record.topic()+",partition"+record.partition() +",offset="+record.offset() + ",value: " + record.value().toString());
        
        // 处理完提交offset
        ack.acknowledge();
    }*/

    @KafkaListener(id="maxwell-container", topics = {"${kafka.consumer.topic}"})
    public void consumerListener(List<ConsumerRecord<?, ?>> consumerRecords, Acknowledgment ack) {
    	for(ConsumerRecord<?, ?> record : consumerRecords) {
            logger.info("【myContainer】kafka consumer消费成功！topic=" + record.topic()+",partition"+record.partition() +",offset="+record.offset() + ",value: " + record.value().toString());
    	}
    	
        // 处理完提交offset
        ack.acknowledge();
    }
    
}
